package org.chen.ctrip.rmq.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.chen.common.ctrip.entity.Product;
import org.chen.ctrip.rmq.domain.AiProcessingMessage;
import org.chen.ctrip.rmq.producer.AiProcessingProducerService;
import org.chen.ctrip.service.ProductService;
import org.chen.ctrip.service.ollama.ItineraryExtractorService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.List;

@Slf4j
@Component
public class AiProcessingMessageListener implements MessageListenerConcurrently {
    
    @Autowired
    private ItineraryExtractorService itineraryExtractorService;
    
    @Autowired
    private ProductService productService;
    
    @Autowired
    private AiProcessingProducerService aiProcessingProducer;
    
    private static final int MAX_RETRY_TIMES = 3;
    
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(
            List<MessageExt> messages,
            ConsumeConcurrentlyContext context) {
        
        for (MessageExt message : messages) {
            try {
                String messageBody = new String(message.getBody(), StandardCharsets.UTF_8);
                AiProcessingMessage aiMessage = AiProcessingMessage.fromJson(messageBody);
                
                if (aiMessage == null) {
                    log.error("解析AI处理消息失败: {}", messageBody);
                    continue; // 跳过无效消息
                }
                
                log.info("开始处理AI消息: productId={}, source={}, traceId={}, reconsumeTimes={}", 
                    aiMessage.getProductId(), aiMessage.getSource(), aiMessage.getTraceId(), 
                    message.getReconsumeTimes());
                
                boolean success = processProductAi(aiMessage);
                
                if (!success) {
                    // 处理失败，检查重试次数
                    if (message.getReconsumeTimes() < MAX_RETRY_TIMES) {
                        log.warn("AI处理失败，消息将重试: productId={}, reconsumeTimes={}", 
                            aiMessage.getProductId(), message.getReconsumeTimes());
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    } else {
                        log.error("AI处理失败，超过最大重试次数: productId={}", aiMessage.getProductId());
                        // 发送到死信队列
                        aiProcessingProducer.sendToDLQ(aiMessage, "超过最大重试次数");
                    }
                }
                
            } catch (Exception e) {
                log.error("处理AI消息异常: msgId={}", message.getMsgId(), e);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
    
    /**
     * 处理单个商品的AI总结
     */
    private boolean processProductAi(AiProcessingMessage message) {
        try {
            String productId = message.getProductId();
            
            Product product = productService.getById(productId);
            if (product == null) {
                log.warn("商品{}不存在，跳过AI处理", productId);
                return true; // 商品不存在认为处理成功
            }
            
            // 防重复处理
            if (Boolean.TRUE.equals(product.getAskAi())) {
                log.info("商品{}已AI处理过，跳过", productId);
                return true;
            }
            
            String pageDiv = product.getPageDiv();
            if (pageDiv == null || pageDiv.trim().isEmpty()) {
                log.warn("商品{}没有页面内容，跳过AI处理", productId);
                return true;
            }
            
            // 记录开始时间
            long startTime = System.currentTimeMillis();

            // 调用AI进行总结
            String firstDayContent = itineraryExtractorService.processFirstDay(pageDiv);
            String lastDayContent = itineraryExtractorService.processLastDay(pageDiv);

            long aiProcessTime = System.currentTimeMillis() - startTime;

            // 更新数据库
            product.setFirstDay(firstDayContent);
            product.setLastDay(lastDayContent);
            product.setAskAi(true);
            product.setUpdateTime(LocalDateTime.now());

            boolean updated = productService.updateById(product);
            if (updated) {
                log.info("商品{}AI总结完成: traceId={}, 耗时={}ms",
                    productId, message.getTraceId(), aiProcessTime);
                return true;
            } else {
                log.error("商品{}AI总结保存失败", productId);
                return false;
            }
            
        } catch (Exception e) {
            log.error("处理商品{}AI总结异常", message.getProductId(), e);
            return false;
        }
    }
}